package edu.hhu.wifispark.task.sparktask;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;
import edu.hhu.wifispark.entry.BasePhone;
import edu.hhu.wifispark.entry.Customerflow;
import edu.hhu.wifispark.entry.Customerlive;
import edu.hhu.wifispark.util.DateUtil;
import edu.hhu.wifispark.util.SparkProUtil;

public class CFlowTaskAll extends Thread implements Serializable{
	private static final long serialVersionUID = 1L;
	public List<BasePhone> bps;
	String taskName= null;
	public CFlowTaskAll(){
		this.taskName = "instoreRate";
	}
	public CFlowTaskAll(long time){
		this.taskName = "instoreRate"+time;
	}
	
	@SuppressWarnings("resource")
	public void run(){
		SparkConf sparkConf = new SparkConf().setAppName(taskName).setMaster(SparkProUtil.MASTER);
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		JavaRDD<BasePhone> phoneRDD = jsc.parallelize(bps);

		
		JavaPairRDD<String, Integer> phonep = phoneRDD.mapToPair(new PairFunction<BasePhone, String, Integer>() {
			private static final long serialVersionUID = 1L;

			public Tuple2<String, Integer> call(BasePhone arg0)
					throws Exception {
				long time = DateUtil.getHour(arg0.getLong("atime"));
				if(arg0.getInt("inner") == 1){
					return new Tuple2<String, Integer>(arg0.getStr("mac")+","+time, 2);
				}else if(arg0.getInt("outer") == 1){
					return new Tuple2<String, Integer>(arg0.getStr("mac")+","+time, 1);
				}else{
					return new Tuple2<String, Integer>(arg0.getStr("mac")+","+time, 0);
				}
			}
		});
		
		JavaPairRDD<String, Integer> res0 = phonep.reduceByKey(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 1L;

			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0>arg1?arg0:arg1;
			}
		});

		JavaPairRDD<String, Integer> res1 = res0.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {
			private static final long serialVersionUID = 1L;

			public Tuple2<String, Integer> call(Tuple2<String, Integer> t)
					throws Exception {
				
				return new Tuple2<String, Integer>(t._1.split(",")[1], t._2);
			}
		});

		Map<Tuple2<String, Integer>, Long> res2 = res1.countByValue();
		Map<Long,Customerflow> flows = new HashMap<Long,Customerflow>();
		for(Tuple2<String, Integer> keys : res2.keySet()){
			System.out.println(keys._1 + (keys._2 == 1?" 门外 ":" 门内 ")+res2.get(keys));
			long time = Long.parseLong(keys._1);
			Customerflow flow = null;
			if(flows.containsKey(time)){
				flow = flows.get(time);
			}else {
				flow = new Customerflow();
				flow.set("atime", time);
				flows.put(time, flow);
			}
			Long num = res2.get(keys);
			if(keys._2 == 1){
				flow.set("outernum", num);
			}else {
				flow.set("innernum", num);
			}
		}
		jsc.stop();
		for(long time : flows.keySet()){
			int oldCount = Customerlive.dao.selectOldCustomer(time);
			flows.get(time).set("oldcustomer", oldCount).save();
		}
	}
	public void setBps(List<BasePhone> ibps){
		this.bps = ibps;
	}
}
